/*
 * sgc_epoll.c
 *
 *  Created on: 2011-7-6
 *      Author: rendong01
 */
#include<stdio.h>
#include<stdlib.h>
#include<string.h>
#include<errno.h>

#include<sys/epoll.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>

#include "../core/sgc_log.h"
#include "sgc_epoll.h"

#define EPOLL_WAIT_TIMEOUT 600

#define EPOLL_CLOSE_CODE(msg) do {\
ev.data.fd=sockfd;\
epoll_ctl(epfd,EPOLL_CTL_DEL,sockfd,&ev);\
close(sockfd);\
config->currentclients--;\
sgc_log_debug("\nclient close fd:%d type:%s\n", sockfd, msg);\
events[i].data.fd=-1;\
config->hander_close(sockfd);\
}while(0);

/**
 * 启动socket服务
 */
void sgc_serversocket_start(sgc_epoll_socketserver_config_t *config)
{
	int epfd, listenfd,connfd,readyfds,i,sockfd, nreadbytes;
	int flag=1,len=sizeof(int);

	struct epoll_event ev, events[config->cfgmaxclients];
	struct sockaddr_in clientaddr;
	struct sockaddr_in serveraddr;
	socklen_t clilen = sizeof(struct sockaddr_in);

	if (config->handler_new == NULL ||
		config->hander_read == NULL){
		sgc_log_error("config error hand is NULL\n");
		exit(1);
	}

	char * buffer = (char *)malloc(config->buffer_size);
	memset(buffer, 0, config->buffer_size);

	epfd = epoll_create(config->cfgmaxclients);
	config->epfd = epfd;

	config->currentclients = 0;
	config->maxclients = 0;

	listenfd = socket(AF_INET, SOCK_STREAM, 0);

	if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &flag, len))
		sgc_log_error("set sockopt SO_REUSEADDR error\n");
	if (setsockopt(listenfd, SOL_SOCKET, SO_SNDBUF, &cfg_sndbuff, len))
		sgc_log_error("set sockopt SO_SNDBUF error\n");
	if (setsockopt(listenfd, SOL_SOCKET, SO_SNDBUF, &cfg_revbuffer, len))
		sgc_log_error("set sockopt SO_SNDBUF error\n");

	setnonblocking(listenfd);

	ev.data.fd = listenfd;
	ev.events=EPOLLIN;

	// 监听端口
	epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
	bzero(&serveraddr, sizeof(serveraddr));
	serveraddr.sin_family = AF_INET;
	char *local_addr= config->local_addr;
	inet_aton(local_addr,&(serveraddr.sin_addr));

	serveraddr.sin_port=htons(config->port);

	if (bind(listenfd,(struct sockaddr *)&serveraddr, sizeof(serveraddr)))
	{
		sgc_log_error("bind fail %s:%d\n", config->local_addr, config->port);
		exit(1);
	}

	sgc_log_debug("start listen %s:%d backlog:%d\n", local_addr, config->port, LISTEN_BACKLOG);
	if (listen(listenfd, LISTEN_BACKLOG))
	{
		sgc_log_error("listen %s:%d max:%d error\n", config->local_addr, config->port, config->cfgmaxclients);
		exit(1);
	}

	for (;;) {
		readyfds=epoll_wait(epfd,events,config->cfgmaxclients,EPOLL_WAIT_TIMEOUT);

		if (readyfds < 0){
			sgc_log_error("epoll_weit errno:%d %s\n", errno, strerror(errno));
		}

		for(i=0;i<readyfds;++i)
		{
			// 新用户连接
			if(events[i].data.fd==listenfd)
			{
				connfd = accept(listenfd, (struct sockaddr *)&clientaddr, &clilen);

				if (config->currentclients >= config->cfgmaxclients){
					sgc_log_debug("clients full,close:%s socket fd:%d clients:%d\n", inet_ntoa(clientaddr.sin_addr), listenfd, config->currentclients);
					close(connfd);
				}else{
					config->currentclients++;
					if (config->currentclients > config->maxclients)
						config->maxclients = config->currentclients;

					if(connfd<0){
						sgc_log_error("accept error connfd<0 fd=%d errno:%d %s\n", connfd, errno, strerror(errno));
						exit(1);
					}
					setnonblocking(connfd);
					ev.data.fd=connfd;
					ev.events=EPOLLIN|EPOLLET;

					sgc_log_debug("new clients:%d, num:%d %s\n\n\n", connfd, config->currentclients, inet_ntoa(clientaddr.sin_addr));
					config->handler_new(connfd, &clientaddr, config);

					epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
				}
			}else if(events[i].events & EPOLLIN)//如果是已经连接的用户，并且收到数据，那么进行读入。
			{
				if ((sockfd = events[i].data.fd) < 0){
					sgc_log_error("EPOLLIN fd < 0\n");
					continue;
				}

				// 一次读完缓冲区
				memset(buffer, 0, config->buffer_size);
				while((nreadbytes = recv(sockfd, buffer, config->buffer_size, 0)) > 0)
				{
					sgc_log_debug("read fd:%d len:%d\n", sockfd, nreadbytes);
					config->hander_read(sockfd, buffer, nreadbytes);

					memset(buffer, 0, config->buffer_size);
				}
				sgc_log_debug("read finish nreadbytes:%d fd:%d errno:%d, err:%s\n", nreadbytes, sockfd, errno, strerror(errno));
				if (nreadbytes < 0){
					if (errno == ECONNRESET){
						EPOLL_CLOSE_CODE("connrset")
					}
				}else if (nreadbytes == 0){
					EPOLL_CLOSE_CODE("normal")
				}
			}else{// 其它异常情况，关闭
				sockfd = events[i].data.fd;
				EPOLL_CLOSE_CODE("exception")
			}
		}
	}
}


/**
 * 设置非阻塞
 */
void setnonblocking(int sock)
{
    int opts;
    opts=fcntl(sock,F_GETFL);
    if(opts<0)
    {
        sgc_log_error("fcntl(sock,F_GETFL) error\n");
        exit(1);
    }

    opts = opts|O_NONBLOCK;

    if(fcntl(sock,F_SETFL,opts)<0)
    {
    	sgc_log_error("fcntl(sock,SETFL,opts)");
        exit(1);
    }
}


ssize_t /* Write "n" bytes to a descriptor */
writen(int fd, const void *buffer, size_t n)
{
	size_t      nleft;
	ssize_t     nwritten;

	nleft = n;
	while (nleft > 0) {
		if ((nwritten = write(fd, buffer, nleft)) < 0) {
			if (errno == EAGAIN || errno == EINTR){// 缓冲区满，继续写入
				usleep(100);
				sgc_log_debug("error writen usleep(100) fd:%d errno:%d error:%s\n", fd, errno, strerror(errno));
				continue;
			}else{
				sgc_log_debug("error writen break fd:%d errno:%d error:%s\n", fd, errno, strerror(errno));
				break;
			}
		}
		if (nwritten == 0){
			sgc_log_debug("error writen break(nwritten=0) fd:%d errno:%d error:%s\n", fd, errno, strerror(errno));
			break;
		}

		nleft -= nwritten;
		buffer   = (char *)buffer + nwritten;
	}
	return(n - nleft);      /* return >= 0 */
}
